package com.hmy.rpc;

import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

/**
 * 1.配置连接工厂<br>
 * 2.建立TCP连接<br>
 * 3.在TCP连接上建立通道<br>
 * 4.定义临时队列replyQueueName，声明唯一标志本次请求corrId，并将replyQueueName和corrId配置要发送的消息队列中<br>
 * 5.使用默认的交换机，发送消息到队列rpc_queue<br>
 * 6.使用阻塞队列BlockingQueue阻塞当前进程<br>
 * 7.收到请求后，将请求放入BlockingQueue中，主进程被唤醒，打印返回类容
 * @author Administrator
 *
 */
public class RPCClient implements AutoCloseable
{
	private Connection connection;
	private Channel channel;
	private String requestQueueName = "rpc_queue";
	private String replyQueueName;

	public RPCClient() throws IOException, TimeoutException
	{
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost("localhost");
		factory.setPort(5682);

		connection = factory.newConnection();
		channel = connection.createChannel();

		// 定义临时队列，并返回生成的队列名
		replyQueueName = channel.queueDeclare().getQueue();
	}

	public String call(String message) throws IOException, InterruptedException
	{
		// 唯一标志本次请求
		final String corrId = UUID.randomUUID().toString(); // 因为是final类型，所以多次发送数据，导致关系ID不变，进而导致服务端返回的数据，客户端不能消费

		// 生成发送消息的属性
		AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().correlationId(corrId).replyTo(replyQueueName)
				.build();

		// 发送消息，发送到默认交换机上
		channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8"));

		// 阻塞队列，用户存储回调结果
		final BlockingQueue<String> response = new ArrayBlockingQueue<String>(1);
		// 服务端的处理返回结果
		channel.basicConsume(replyQueueName, true, new DefaultConsumer(channel)
		{
			// 这里的变量都是final类型
			@Override
			public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
					byte[] body) throws IOException
			{
				System.out.println("properties.getCorrelationId():" + properties.getCorrelationId() + "--corrId:" + corrId);
				if (properties.getCorrelationId().equals(corrId))
				{
					response.offer(new String(body, "UTF-8"));
				}
			}
		});

		// 获取服务端的处理返回结果
		return response.take();
	}

	@Override
	public void close() throws Exception
	{
		connection.close();
	}

	public static void main(String[] args)
	{
		try
		{
			@SuppressWarnings("resource")
			RPCClient fibonacciRpc = new RPCClient();
			
//			System.out.println(" [x] Requesting fib(2)");
//			final String response = fibonacciRpc.call("2");
//			System.out.println(" [.] Got '" + response + "'");
			
			int i = 2;
			while (true)
			{
				Thread.sleep(1000);
				System.out.println(" [x] Requesting fib(" + i + ")");
				final String response = fibonacciRpc.call(i + "");
				System.out.println(" [.] Got '" + response + "'");
				i++;
			}

		} catch (IOException e)
		{
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (TimeoutException e)
		{
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (InterruptedException e)
		{
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

}
